package avicit.bdp.dcs.json.service.reader;

import avicit.bdp.common.service.service.DataSourceService;
import avicit.bdp.common.utils.database.ColumnInfo;
import avicit.bdp.common.utils.database.DBUtils;
import avicit.bdp.core.constant.Constants;
import avicit.bdp.core.util.json.JSONUtils;
import avicit.bdp.dcs.datasource.dto.JobDatasource;
import avicit.bdp.dcs.datasource.service.DataSourceUtils;
import avicit.bdp.dcs.datasource.service.DsDatasourceService;
import avicit.bdp.dcs.job.dto.JobDTO;
import avicit.bdp.dcs.json.service.util.BuildJsonUtil;
import avicit.bdp.dcs.plugin.BaseDataxParam;
import avicit.bdp.dcs.plugin.rdbms.param.RdbmsDataxParam;
import avicit.bdp.dcs.task.dto.ColumnMappingDto;
import avicit.bdp.dcs.task.dto.TaskDTO;
import avicit.bdp.dcs.tools.enums.SyncCondition;
import avicit.bdp.dcs.tools.enums.SyncMode;
import avicit.bdp.dcs.utils.DataTypeConstants;
import avicit.bdp.dcs.utils.JdbcConstants;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.google.common.collect.Lists;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;

@Component
public class RdbmsReaderBuilder {

  @Autowired private DataSourceService dataSourceService;
  @Autowired private DataSourceUtils dataSourceUtils;
  @Autowired private DsDatasourceService dsDatasourceService;
  @Autowired private BuildJsonUtil buildJsonUtil;

  private static final String SRC_COLUMN_NAME = "srcColumnName";

  /** 构建结构化采集Reader */
  public BaseDataxParam buildRdbmsReaderByStructure(TaskDTO taskDTO, JobDTO jobDTO)
          throws Exception {
    String taskId = taskDTO.getId();

    JobDatasource readerDatasource = dataSourceUtils.convertDatasource(jobDTO.getSrcDatasourceId());
    String schema = dataSourceUtils.getDbSchema(readerDatasource);
    String tableName = DBUtils.convertTableName(readerDatasource.getDbType(), taskDTO.getSrcTableName());
    tableName = dataSourceUtils.appendSchema2TableName(readerDatasource.getDbType(), tableName, schema);

    RdbmsDataxParam dataxParam = new RdbmsDataxParam();
    dataxParam.setTableList(Lists.newArrayList(tableName));
    dataxParam.setColumnList(getReaderColumns(taskDTO, readerDatasource.getDbType()));

    dataxParam.setSplitPk(getPrimaryKey(taskId, jobDTO));
    dataxParam.setWhereParam(buildWhereCondition(taskDTO, jobDTO));
    dataxParam.setQuerySql(Constants.STRING_BLANK);
    return dataxParam;
  }

  /** 构建where条件 */
  private String buildWhereCondition(TaskDTO taskDTO, JobDTO jobDTO) {
    StringBuilder sb = new StringBuilder();
    if (taskDTO.getSyncMode() != null
            && taskDTO.getSyncMode() == SyncMode.DATAX_INCREMENT_SYNC.getCode()) {

      // 同步条件字段，时间戳字段：1，主键字段：2
      if (taskDTO.getSyncConditionField() != null
              && taskDTO.getSyncConditionField() == SyncCondition.TIME.getCode()) {
        String columnType = getSrcColumnType(taskDTO.getMappingColumn(), taskDTO.getFieldName());
        sb.append(dataSourceService.getStartDateFunction(jobDTO.getSrcDatasourceId(), columnType))
                .append("<=")
                .append(taskDTO.getFieldName())
                .append(Constants.SPACE)
                .append(Constants.AND)
                .append(Constants.SPACE)
                .append(taskDTO.getFieldName())
                .append("<")
                .append(dataSourceService.getEndDateFunction(jobDTO.getSrcDatasourceId(), columnType));
      }
    }

    // 添加自定义过滤条件
    if (StringUtils.isNotEmpty(taskDTO.getFilterCondition())) {
      if (StringUtils.isNotEmpty(sb.toString())) {
        sb.append(Constants.SPACE).append(Constants.AND).append(Constants.SPACE);
      }
      sb.append(taskDTO.getFilterCondition());
    }
    return sb.toString();
  }

  private List<String> getReaderColumns(TaskDTO taskDTO, String dbType) {
    List<String> readerColumns = new ArrayList<>();
    JSONArray jsonArray = JSON.parseArray(taskDTO.getMappingColumn());
    for (int i = 0; i < jsonArray.size(); i++) {
      JSONObject jsonObject = jsonArray.getJSONObject(i);
      readerColumns.add(
              buildJsonUtil.convertColumnName(dbType, jsonObject.getString(SRC_COLUMN_NAME)));
    }

    return readerColumns;
  }

  /** 获取主键列 */
  public String getPrimaryKey(String taskId, JobDTO jobDTO) throws Exception {
    // 获取主键列
    // DataX 仅支持切分主键为一个,并且类型为整数或者字符串类型
    String primaryKey = Constants.STRING_BLANK;
    JobDatasource writerDatasource = dataSourceUtils.convertDatasource(jobDTO.getDstDatasourceId());
    if (JdbcConstants.HIVE.equals(writerDatasource.getDbType())) {
      return primaryKey;
    }

    List<ColumnInfo> primaryKeyList = dsDatasourceService.getPrimaryKeyList(taskId);
    if (CollectionUtils.isNotEmpty(primaryKeyList) && primaryKeyList.size() == 1) {
      ColumnInfo pkColumn = primaryKeyList.get(0);
      if (checkPkIntType(pkColumn.getType())) {
        primaryKey = pkColumn.getName();
      }
    }

    return primaryKey;
  }

  /** 检查主键类型仅为整数 */
  private boolean checkPkIntType(String dataType) {
    boolean result;
    dataType = dataType.toLowerCase();
    switch (dataType) {
      case DataTypeConstants.TINYINT:
      case DataTypeConstants.SMALLINT:
      case DataTypeConstants.MEDIUMINT:
      case DataTypeConstants.INT:
      case DataTypeConstants.BIGINT:
        result = true;
        break;
      default:
        result = false;
        break;
    }

    if (dataType.contains(DataTypeConstants.SMALLINT)
            || dataType.contains(DataTypeConstants.INTEGER)
            || dataType.contains(DataTypeConstants.BIGINT)) {
      result = true;
    }
    return result;
  }

  /**
   * 获取源表字段类型
   * @param mappingColumn
   * @return
   */
  private String getSrcColumnType(String mappingColumn, String fieldName) {
    String srcColumnType = null;
    List<ColumnMappingDto> columnInfoList = JSONUtils.toList(mappingColumn, ColumnMappingDto.class);
    if (CollectionUtils.isNotEmpty(columnInfoList)) {
      for(ColumnMappingDto dto : columnInfoList){
        if(fieldName.equalsIgnoreCase(dto.getSrcColumnName())){
          srcColumnType = dto.getSrcColumnType();
          break;
        }
      }
    }
    return srcColumnType;
  }

}
